# -*- coding: utf-8 -*-
# @Time    : 2025/8/5 15:18
# @Author  : Zhaoxu
# @File    : together.py
# @Software: PyCharm

# -*- coding:utf-8 -*-
"""
Realtime Voice Chat  (Mic → XF ASR → GPT → XF TTS → PyAudio 播放)
-----------------------------------------------------------------
依赖:
    pip install openai websocket-client pyaudio
    # Windows 安装 PyAudio 可先:
    #   pip install pipwin
    #   pipwin install pyaudio
-----------------------------------------------------------------
用法:
    1. 麦克风说话 (最长 10 秒；静音 2.5 秒自动结束)
    2. 讯飞实时语音转写得到文本
    3. 交给 GPT 生成回答 (DeepSeek Chat 示例)
    4. GPT 答案边生成、边切块、边走讯飞 TTS 合成
    5. 每帧音频立即写入 PyAudio, 实时播放
"""
import queue
import threading
import time
import os
import base64
import json
import hashlib
import hmac
from urllib.parse import urlencode, quote
from datetime import datetime
from wsgiref.handlers import format_date_time

import websocket
import pyaudio
from websocket import create_connection
from openai import OpenAI

# ========= GPT → 队列 ========= #
def gpt_stream_to_queue(q: queue.Queue, user_prompt: str):
    """
    GPT 流式输出 → 实时推送 → 入队
    优化：每5个token或遇到标点就推送，实现更流畅的实时对话
    """
    client = OpenAI(
        api_key="sk-9bc1ce71791b49f083e625f2f2788b2d",
        base_url="https://api.deepseek.com"
    )
    resp = client.chat.completions.create(
        model="deepseek-chat",
        messages=[
            {"role": "system", "content": (
                "You are a helpful assistant, your name is xiaoxi or 小西. "
                "If user speaks English, always reply in fluent English. "
                "If user speaks Chinese, reply in Chinese. "
                "Respond naturally and conversationally. "
                "Keep responses concise and engaging."
            )},
            {"role": "user", "content": user_prompt}
        ],
        stream=True
    )

    buffer = ""
    token_count = 0
    for part in resp:
        delta = part.choices[0].delta
        if not getattr(delta, "content", None):
            continue
        token = delta.content
        print(token, end="", flush=True)          # 在终端同步显示
        buffer += token
        token_count += 1
        
        # 优化：每5个token或遇到标点就推送，实现更流畅的实时体验
        if (token_count >= 5 or 
            buffer.endswith(('.', '!', '?', '。', '！', '？', ',', '，', ';', '；')) or
            buffer.endswith((' ', '  '))):  # 空格也作为分割点
            if buffer.strip():
                q.put(buffer.strip())
                buffer = ""
                token_count = 0
    
    # 推送剩余内容
    if buffer.strip():
        q.put(buffer.strip())
    q.put(None)     # 结束信号


# ========= 讯飞 TTS websocket ========= #
class Ws_Param:
    def __init__(self, appid, apikey, apisecret, text):
        self.APPID = appid
        self.APIKey = apikey
        self.APISecret = apisecret
        self.Text = text
        self.CommonArgs = {"app_id": self.APPID}
        self.BusinessArgs = {"aue": "raw",
                             "auf": "audio/L16;rate=16000",
                             "vcn": "x4_yezi",
                             "tte": "utf8"}
        self.Data = {"status": 2,
                     "text": str(base64.b64encode(self.Text.encode('utf-8')), "utf8")}

    def create_url(self):
        url = "wss://tts-api.xfyun.cn/v2/tts"
        now = datetime.now()
        date = format_date_time(time.mktime(now.timetuple()))
        sig_origin = f"host: ws-api.xfyun.cn\n" \
                     f"date: {date}\n" \
                     f"GET /v2/tts HTTP/1.1"
        sign = hmac.new(self.APISecret.encode(), sig_origin.encode(),
                        digestmod=hashlib.sha256).digest()
        sign = base64.b64encode(sign).decode()
        auth_origin = f'api_key="{self.APIKey}", algorithm="hmac-sha256", ' \
                      f'headers="host date request-line", signature="{sign}"'
        authorization = base64.b64encode(auth_origin.encode()).decode()
        return url + "?" + urlencode({"authorization": authorization,
                                      "date": date,
                                      "host": "ws-api.xfyun.cn"})


def tts_stream_and_play(appid, apikey, apisecret, text):
    """
    对 text 进行 TTS；收到音频帧立即写 PyAudio，阻塞直到播放完
    优化：更流畅的音频播放体验
    """
    if not text.strip():
        return
        
    p = pyaudio.PyAudio()
    stream = p.open(format=pyaudio.paInt16,
                    channels=1,
                    rate=16000,
                    output=True)

    done_flag = threading.Event()

    def on_message(ws, message):
        msg = json.loads(message)
        audio_bytes = base64.b64decode(msg["data"]["audio"])
        stream.write(audio_bytes)                  # 实时播放
        if msg["data"]["status"] == 2:
            # TTS合成完成，设置完成标志但不主动关闭连接
            done_flag.set()

    def on_error(ws, error):
        print(f"❌ TTS错误: {error}")

    def on_close(ws, close_status_code, close_msg):
        # 只在连接关闭时清理音频资源
        stream.stop_stream()
        stream.close()
        p.terminate()
        done_flag.set()

    def on_open(ws):
        def run(*_):
            req = {"common": wsParam.CommonArgs,
                   "business": wsParam.BusinessArgs,
                   "data": wsParam.Data}
            ws.send(json.dumps(req))
        threading.Thread(target=run).start()

    wsParam = Ws_Param(appid, apikey, apisecret, text)
    websocket.enableTrace(False)
    ws = websocket.WebSocketApp(wsParam.create_url(),
                                on_open=on_open,
                                on_message=on_message,
                                on_error=on_error,
                                on_close=on_close)
    ws.run_forever(sslopt={"cert_reqs": 0})
    done_flag.wait()           # 等播放完毕


def consumer_tts_play(q: queue.Queue, appid, apikey, apisecret):
    """
    消费队列并播放TTS
    优化：更快的响应和更流畅的播放体验
    """
    idx = 1
    while True:
        txt = q.get()
        if txt is None:
            break
        print(f"\n🔊 播放第 {idx} 段（{len(txt)} 字符）...")
        tts_stream_and_play(appid, apikey, apisecret, txt)
        idx += 1
    print("\n✅ GPT 回答播放完成")


# ========= 讯飞实时 ASR ========= #
def gen_signa(appid, apikey, ts):
    base = appid + ts
    md5 = hashlib.md5(base.encode()).hexdigest()
    signa = hmac.new(apikey.encode(), md5.encode(), digestmod='sha1').digest()
    return base64.b64encode(signa).decode()


class XFRealtimeASR:
    def __init__(self, appid, apikey, vad_eos_ms=1500):
        self.appid = appid
        self.apikey = apikey
        self.vad_eos_ms = vad_eos_ms
        self.ws = None
        self.handshake_done = threading.Event()
        self.closed = False
        self.end_tag = "{\"end\":true}"
        self.result_text = ""
        self.last_result_time = None
        self.record_start_time = None

    def connect(self):
        ts = str(int(time.time()))
        signa = gen_signa(self.appid, self.apikey, ts)
        url = f"ws://rtasr.xfyun.cn/v1/ws?appid={self.appid}&ts={ts}&signa={quote(signa)}&vad_eos={self.vad_eos_ms}"
        print("🔗 连接语音识别服务...")
        self.ws = create_connection(url)
        threading.Thread(target=self._recv_thread_func, daemon=True).start()

    def start_mic(self):
        """
        开始录音 + 实时识别，返回整句文本
        优化：更好的英文识别体验
        """
        if not self.handshake_done.wait(10):
            print("❌ 握手失败")
            return ""
        CHUNK = 640
        FORMAT = pyaudio.paInt16
        CHANNELS = 1
        RATE = 16000
        pa = pyaudio.PyAudio()
        stream = pa.open(format=FORMAT, channels=CHANNELS, rate=RATE,
                         input=True, frames_per_buffer=CHUNK)
        print("🎤 开始说话（支持中英文，静音2.5秒或10秒自动结束）")

        try:
            while not self.closed:
                data = stream.read(CHUNK, exception_on_overflow=False)
                self.ws.send(data, opcode=0x2)

                if self.record_start_time:
                    elapsed = time.time() - self.record_start_time
                    if elapsed >= 10 or (time.time() - self.last_result_time) > (self.vad_eos_ms / 1000):
                        print("✅ 语音识别结束")
                        self.ws.send(self.end_tag.encode())
                        break
                time.sleep(0.04)
        except KeyboardInterrupt:
            print("⏹️ 用户中断")
        finally:
            stream.stop_stream()
            stream.close()
            pa.terminate()
            time.sleep(0.2)
            return self.result_text

    def parse_data_text(self, data_str):
        try:
            data_dict = json.loads(data_str)
            words = []
            for rt in data_dict.get('cn', {}).get('st', {}).get('rt', []):
                for ws in rt['ws']:
                    for cw in ws['cw']:
                        words.append(cw['w'])
            return ''.join(words)
        except Exception as e:
            print(f"❌ 解析语音识别文本出错: {e}")
            return ""

    def _recv_thread_func(self):
        try:
            while self.ws.connected:
                result = self.ws.recv()
                if not result:
                    break
                j = json.loads(result)
                if j["action"] == "started":
                    self.handshake_done.set()
                elif j["action"] == "result":
                    txt = self.parse_data_text(j.get("data", ""))
                    if txt:
                        print(f"📝 识别: {txt}")
                        self.result_text += txt
                        self.last_result_time = time.time()
                        if self.record_start_time is None:
                            self.record_start_time = time.time()
                elif j["action"] == "error":
                    print(f"❌ 语音识别错误: {j}")
                    self.closed = True
                    break
        except Exception as e:
            if not self.closed:
                print(f"❌ 语音识别接收线程异常: {e}")

    def close(self):
        """关闭连接 - 在程序结束时调用"""
        self.closed = True
        try:
            if self.ws:
                self.ws.close()
                print("🔌 语音识别连接已关闭")
        except:
            pass


# ========= main ========= #
def main():
    """
    主函数 - 实时语音对话系统
    优化：更流畅的英文对话体验
    """
    print("=" * 60)
    print("🤖 实时语音对话系统 v2.0")
    print("💬 支持中英文实时对话")
    print("🎤 说话后自动识别并回答")
    print("🔊 AI回答实时语音播放")
    print("=" * 60)
    
    # ======= 讯飞账号配置 =======
    XF_APPID_ASR   = "9646138c"
    XF_APIKEY_ASR  = "82adf64eb83cf7b9ec6f4ae215943e0d"
    XF_APPID_TTS   = "9646138c"
    XF_APIKEY_TTS  = "ea89fb89877aded124af01b32cb14324"
    XF_SECRET_TTS  = "MzlhYmQyMjI5ODYzYzU2ODM2YWMxYWNk"
    
    asr = None
    try:
        # ---- Step1: 语音实时转写 ---- #
        print("\n🎤 开始语音识别...")
        asr = XFRealtimeASR(XF_APPID_ASR, XF_APIKEY_ASR, vad_eos_ms=2500)
        asr.connect()
        user_text = asr.start_mic()
        print(f"\n📝 识别结果: {user_text}")

        if not user_text.strip():
            print("⚠️ 未识别到文本，程序结束")
            return

        # ---- Step2: GPT → TTS → 播放 ---- #
        print("\n🧠 AI思考中...")
        q = queue.Queue()
        threading.Thread(target=gpt_stream_to_queue,
                         args=(q, user_text), daemon=True).start()
        consumer_tts_play(q, XF_APPID_TTS, XF_APIKEY_TTS, XF_SECRET_TTS)
        
        print("\n✅ 对话完成！")
        
    except KeyboardInterrupt:
        print("\n👋 用户中断，程序退出")
    except Exception as e:
        print(f"\n❌ 程序出错: {e}")
        print("请检查网络连接和API配置")
    finally:
        # 程序结束时统一关闭连接
        if asr:
            asr.close()
        print("\n🔌 所有连接已关闭")


if __name__ == "__main__":
    main()
